/*
 * Copyright (c) 2018 Baidu, Inc. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.baidubce.formula.consul.config.spring.boot;

import com.baidubce.formula.consul.config.spring.boot.auth.BmsAuthClient;
import com.ecwid.consul.v1.ConsistencyMode;
import com.ecwid.consul.v1.ConsulClient;
import com.ecwid.consul.v1.OperationException;
import com.ecwid.consul.v1.QueryParams;
import com.ecwid.consul.v1.Response;
import com.ecwid.consul.v1.kv.model.GetValue;
import io.micrometer.core.annotation.Timed;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.endpoint.event.RefreshEvent;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.context.SmartLifecycle;
import org.springframework.core.style.ToStringCreator;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;

import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @author luoguangming
 */
public class ConfigWatch implements ApplicationEventPublisherAware, SmartLifecycle {

    private static final Logger logger = LoggerFactory.getLogger(ConfigWatch.class);

    private final ConsulConfigProperties properties;

    private final ConsulClient consul;

    private final List<ThreadPoolTaskScheduler> taskSchedulers;

    private final AtomicBoolean running = new AtomicBoolean(false);

    private LinkedHashMap<String, Long> consulIndexes;

    private ApplicationEventPublisher publisher;

    private boolean firstTime = true;

    private ScheduledFuture<?>[] watchFutures;

    private BmsAuthClient bmsAuthClient;

    public ConfigWatch(ConsulConfigProperties properties, ConsulClient consul, BmsAuthClient bmsAuthClient,
                       LinkedHashMap<String, Long> initialIndexes) {
        this(properties, consul, bmsAuthClient, initialIndexes, null);
    }

    public ConfigWatch(ConsulConfigProperties properties, ConsulClient consul, BmsAuthClient bmsAuthClient,
                       LinkedHashMap<String, Long> initialIndexes, List<ThreadPoolTaskScheduler> taskSchedulers) {
        this.properties = properties;
        this.consul = consul;
        this.consulIndexes = new LinkedHashMap<>(initialIndexes);
        this.watchFutures = new ScheduledFuture<?>[consulIndexes.size()];
        this.bmsAuthClient = bmsAuthClient;
        if (taskSchedulers == null) {
            this.taskSchedulers = getTaskSchedulers();
        } else {
            this.taskSchedulers = taskSchedulers;
        }
    }

    private List<ThreadPoolTaskScheduler> getTaskSchedulers() {
        List<ThreadPoolTaskScheduler> taskSchedulers = new ArrayList<>();
        Object[] keys = consulIndexes.keySet().toArray();
        for (int i = 0; i < consulIndexes.size(); i++) {
            ThreadPoolTaskScheduler threadPoolTaskScheduler = getTaskScheduler();
            if (!StringUtils.isEmpty(keys[i])) {
                String[] keyArray = keys[i].toString().split("/");
                if (keyArray.length >= 1) {
                    keys[i] = keyArray[keyArray.length - 1];
                    threadPoolTaskScheduler.setThreadNamePrefix(keys[i] + "-");
                }
            }
            taskSchedulers.add(threadPoolTaskScheduler);
        }
        return taskSchedulers;
    }

    private ThreadPoolTaskScheduler getTaskScheduler() {
        ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
        taskScheduler.initialize();
        return taskScheduler;
    }

    @Override
    public void setApplicationEventPublisher(ApplicationEventPublisher publisher) {
        this.publisher = publisher;
    }

    @Override
    public void start() {
        if (this.running.compareAndSet(false, true)) {
            int i = 0;
            for (String context : this.consulIndexes.keySet()) {
                watchFutures[i] = taskSchedulers.get(i).scheduleWithFixedDelay(
                        () -> watchConfigKeyValues(context), this.properties.getWatch().getDelay());
                i++;
            }
        }
    }

    @Override
    public boolean isAutoStartup() {
        return true;
    }

    @Override
    public void stop(Runnable callback) {
        this.stop();
        callback.run();
    }

    @Override
    public int getPhase() {
        return 0;
    }

    @Override
    public void stop() {
        if (this.running.compareAndSet(true, false) && this.watchFutures != null) {
            for (int i = 0; i < watchFutures.length; i++) {
                watchFutures[i].cancel(true);
            }
        }
    }

    @Override
    public boolean isRunning() {
        return this.running.get();
    }

    @Timed("consul.watch-config-keys")
    public void watchConfigKeyValues(String context) {
        if (this.running.get()) {
            if (this.properties.getFormat() != Format.FILES && !context.endsWith("/")) {
                context = context + "/";
            }

            try {
                Long currentIndex = this.consulIndexes.get(context);
                if (currentIndex == null) {
                    currentIndex = -1L;
                }

                logger.debug("watching consul for context '" + context + "' with index " + currentIndex);

                if (properties.isTokenEnabled() && StringUtils.isEmpty(bmsAuthClient.getToken())) {
                    bmsAuthClient.getTokenFromServer(properties.getAuthUri());
                }

                QueryParams queryParams = QueryParams.Builder.builder()
                        .setConsistencyMode(ConsistencyMode.STALE)
                        .setIndex(currentIndex)
                        .setWaitTime(this.properties.getWatch().getWaitTime())
                        .build();
                Response<List<GetValue>> response = this.consul.getKVValues(context, bmsAuthClient.getToken(),
                        queryParams);

                // if response.value == null, response was a 404, otherwise it was a 200
                // reducing churn if there wasn't anything
                if (response.getValue() != null && !response.getValue().isEmpty()) {
                    Long newIndex = response.getConsulIndex();

                    if (newIndex != null && !newIndex.equals(currentIndex)) {
                        // don't publish the same index again, don't publish the first
                        // time (-1) so index can be primed
                        if (!this.consulIndexes.containsValue(newIndex) && !currentIndex.equals(-1L)) {
                            logger.trace("Context " + context + " has new index " + newIndex);
                            RefreshEventData data = new RefreshEventData(context, currentIndex, newIndex);
                            this.publisher.publishEvent(new RefreshEvent(this, data, data.toString()));
                        } else if (logger.isTraceEnabled()) {
                            logger.trace("Event for index already published for context " + context);
                        }
                        this.consulIndexes.put(context, newIndex);
                    } else if (logger.isTraceEnabled()) {
                        logger.trace("Same index for context " + context);
                    }
                } else if (logger.isTraceEnabled()) {
                    logger.trace("No value for context " + context);
                }

            } catch (Exception e) {
                // OperationException(statusCode=403, statusMessage='Forbidden', statusContent='ACL not found')
                if (e instanceof OperationException && ((OperationException) e).getStatusCode() == 403) {
                    logger.info("Token has expired, try to get new token from server.");
                    String newToken = bmsAuthClient.getTokenFromServer(this.properties.getAuthUri());
                    logger.info("New Token is: " + newToken);
                }
                // only fail fast on the initial query, otherwise just log the error
                if (this.firstTime && this.properties.isFailFast()) {
                    logger.error(
                            "Fail fast is set and there was an error reading configuration from consul.");
                    ReflectionUtils.rethrowRuntimeException(e);
                } else if (logger.isTraceEnabled()) {
                    logger.trace("Error querying consul Key/Values for context '"
                            + context + "'", e);
                } else if (logger.isWarnEnabled()) {
                    // simplified one line log message in the event of an agent failure
                    logger.warn("Error querying consul Key/Values for context '"
                            + context + "'. Message: " + e.getMessage());
                }
            }
        }
        this.firstTime = false;
    }

    static class RefreshEventData {

        private final String context;

        private final Long prevIndex;

        private final Long newIndex;

        RefreshEventData(String context, Long prevIndex, Long newIndex) {
            this.context = context;
            this.prevIndex = prevIndex;
            this.newIndex = newIndex;
        }

        public String getContext() {
            return this.context;
        }

        public Long getPrevIndex() {
            return this.prevIndex;
        }

        public Long getNewIndex() {
            return this.newIndex;
        }

        @Override
        public boolean equals(Object o) {
            if (this == o) {
                return true;
            }
            if (o == null || getClass() != o.getClass()) {
                return false;
            }
            RefreshEventData that = (RefreshEventData) o;
            return Objects.equals(this.context, that.context)
                    && Objects.equals(this.prevIndex, that.prevIndex)
                    && Objects.equals(this.newIndex, that.newIndex);
        }

        @Override
        public int hashCode() {
            return Objects.hash(this.context, this.prevIndex, this.newIndex);
        }

        @Override
        public String toString() {
            return new ToStringCreator(this).append("context", this.context)
                    .append("prevIndex", this.prevIndex).append("newIndex", this.newIndex)
                    .toString();
        }

    }

}
